package org.ngbx.demo.flink.clickevent.statistics;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.ngbx.demo.flink.clickevent.UserClickedEvent;
import org.ngbx.demo.flink.clickevent.statistics.event.UserClickStatisticsPeriodicallyCompletedEvent;
import org.ngbx.demo.flink.clickevent.statistics.event.UserClickStatisticsPeriodicallyCompletedEventSerializationSchema;
import org.ngbx.demo.flink.clickevent.statistics.event.UserClickedEventDeserializationSchema;
import org.ngbx.demo.flink.clickevent.statistics.functions.UserClickStatisticsPeriodicallyCompletedEventProcessor;
import org.ngbx.demo.flink.clickevent.statistics.functions.UserClickedEventCountingAggregator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class UserClickedEventStatisticsJob {
    private static final Logger log = LoggerFactory.getLogger(UserClickedEventStatisticsJob.class);

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建kafka配置属性
        Properties kafkaProps = createKafkaProperties();

        KafkaSource<UserClickedEvent> source = KafkaSource.<UserClickedEvent>builder()
                .setProperties(kafkaProps)
                .setTopics("user_clicked_event")
                .setGroupId("user-clicked-event-statistic")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new UserClickedEventDeserializationSchema())
                .build();

        WatermarkStrategy<UserClickedEvent> watermarkStrategy = WatermarkStrategy
                .<UserClickedEvent>forBoundedOutOfOrderness(Duration.ofMillis(200))
                .withTimestampAssigner((clickEvent, l) -> clickEvent.getOccurrenceTime());
        DataStream<UserClickedEvent> userClickedEventDS = env.fromSource(source, watermarkStrategy, "UserClickedEvent Source");
        //userClickedEventDS.print();
        // 设定统计窗口为20秒
        WindowAssigner<Object, TimeWindow> winAssigner = TumblingProcessingTimeWindows.of(Time.of(20, TimeUnit.SECONDS));

        DataStream<UserClickStatisticsPeriodicallyCompletedEvent> statisticsDS = userClickedEventDS
                .keyBy(UserClickedEvent::getUserId)
                .window(winAssigner)
                .aggregate(new UserClickedEventCountingAggregator(),
                        new UserClickStatisticsPeriodicallyCompletedEventProcessor())
                .name("UserClickedEvent Counter");

        statisticsDS.sinkTo(
                KafkaSink.<UserClickStatisticsPeriodicallyCompletedEvent>builder()
                        .setBootstrapServers(kafkaProps.getProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))
                        .setRecordSerializer(
                                KafkaRecordSerializationSchema.builder()
                                        .setTopic("user_click_statistics_periodically_completed_event")
                                        .setValueSerializationSchema(new UserClickStatisticsPeriodicallyCompletedEventSerializationSchema())
                                        .build()
                        ).setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                        .build()
        ).name("ClickEventStatistics Sink");

        env.execute("UserClickedEvent Statistics");
    }

    private static Properties createKafkaProperties() {
        Properties kafkaProps = new Properties();
        kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringDeserializer.class.getCanonicalName());
        // kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, UserClickedEventDeserializationSchema.class.getCanonicalName());
        return kafkaProps;
    }
}
